package com.ycy;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;

import java.io.IOException;
import java.util.NoSuchElementException;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;

/**
 *
 */
public class DistributedQueueTest {


  public ZooKeeper createClient() {
    String url = "10.211.55.3:2184,10.211.55.3:2183,10.211.55.3:2182";
    try {
      return new ZooKeeper(url, 120000, new Watcher() {
        @Override
        public void process(WatchedEvent event) {

          System.out.println(event);
        }
      });
    } catch (IOException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Test
  public void testOffer1() throws Exception {
    String dir = "/ycy/queue";
    String testString = "Hello World";
    final int numClients = 1;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    for (int i = 0; i < 100; i++) {

      queueHandles[0].offer(testString.getBytes());
    }

//    byte[] dequeuedBytes = queueHandles[0].remove();

//    assertEquals(new String(dequeuedBytes), testString);
  }

  @Test
  public void testOffer2() throws Exception {
    String dir = "/testOffer2";
    String testString = "Hello World";
    final int numClients = 2;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    queueHandles[0].offer(testString.getBytes());

    byte[] dequeuedBytes = queueHandles[1].remove();
    assertEquals(new String(dequeuedBytes), testString);
  }

  @Test
  public void testTake1() throws Exception {
    String dir = "/testTake1";
    String testString = "Hello World";
    final int numClients = 1;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    queueHandles[0].offer(testString.getBytes());

    byte[] dequeuedBytes = queueHandles[0].take();
    assertEquals(new String(dequeuedBytes), testString);
  }

  @Test
  public void testRemove1() throws Exception {
    String dir = "/testRemove1";
    final int numClients = 1;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    try {
      queueHandles[0].remove();
    } catch (NoSuchElementException e) {
      return;
    }

//    fail();fail
  }

  public void createNremoveMtest(String dir, int n, int m) throws Exception {
    String testString = "Hello World";
    final int numClients = 2;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    for (int i = 0; i < n; i++) {
      String offerString = testString + i;
      queueHandles[0].offer(offerString.getBytes());
    }

    byte[] data = null;
    for (int i = 0; i < m; i++) {
      data = queueHandles[1].remove();
    }

    assertNotNull(data);
    assertEquals(new String(data), testString + (m - 1));
  }

  @Test
  public void testRemove2() throws Exception {
    createNremoveMtest("/ycy/queue1", 1, 4);
  }

  @Test
  public void testRemove3() throws Exception {
    createNremoveMtest("/ycy/queue", 1000, 1000);
  }

  public void createNremoveMelementTest(String dir, int n, int m) throws Exception {
    String testString = "Hello World";
    final int numClients = 2;
    ZooKeeper[] clients = new ZooKeeper[numClients];
    DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    for (int i = 0; i < n; i++) {
      String offerString = testString + i;
      queueHandles[0].offer(offerString.getBytes());
    }

    for (int i = 0; i < m; i++) {
      queueHandles[1].remove();
    }
    assertEquals(new String(queueHandles[1].element()), testString + m);
  }

  @Test
  public void testElement1() throws Exception {
    createNremoveMelementTest("/testElement1", 1, 0);
  }

  @Test
  public void testElement2() throws Exception {
    createNremoveMelementTest("/testElement2", 10, 2);
  }

  @Test
  public void testElement3() throws Exception {
    createNremoveMelementTest("/testElement3", 1000, 500);
  }

  @Test
  public void testElement4() throws Exception {
    createNremoveMelementTest("/testElement4", 1000, 1000 - 1);
  }

  @Test
  public void testTakeWait1() throws Exception {
    String dir = "/testTakeWait1";
    final String testString = "Hello World";
    final int numClients = 1;
    final ZooKeeper[] clients = new ZooKeeper[numClients];
    final DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }

    final byte[][] takeResult = new byte[1][];
    Thread takeThread = new Thread(() -> {
      try {
        takeResult[0] = queueHandles[0].take();
      } catch (KeeperException | InterruptedException ignore) {
        // no op
      }
    });
    takeThread.start();

    Thread.sleep(1000);
    Thread offerThread = new Thread(() -> {
      try {
        queueHandles[0].offer(testString.getBytes());
      } catch (KeeperException | InterruptedException ignore) {
        // no op
      }
    });
    offerThread.start();
    offerThread.join();

    takeThread.join();

    assertNotNull(takeResult[0]);
    assertEquals(new String(takeResult[0]), testString);
  }

  @Test
  public void testTakeWait2() throws Exception {
    String dir = "/testTakeWait2";
    final String testString = "Hello World";
    final int numClients = 1;
    final ZooKeeper[] clients = new ZooKeeper[numClients];
    final DistributedQueue[] queueHandles = new DistributedQueue[numClients];
    for (int i = 0; i < clients.length; i++) {
      clients[i] = createClient();
      queueHandles[i] = new DistributedQueue(clients[i], dir, null);
    }
    int numAttempts = 2;
    for (int i = 0; i < numAttempts; i++) {
      final byte[][] takeResult = new byte[1][];
      final String threadTestString = testString + i;
      Thread takeThread = new Thread(() -> {
        try {
          takeResult[0] = queueHandles[0].take();
        } catch (KeeperException | InterruptedException ignore) {
          // no op
        }
      });
      takeThread.start();

      Thread.sleep(1000);
      Thread offerThread = new Thread(() -> {
        try {
          queueHandles[0].offer(threadTestString.getBytes());
        } catch (KeeperException | InterruptedException ignore) {
          // no op
        }
      });
      offerThread.start();
      offerThread.join();

      takeThread.join();

      assertNotNull(takeResult[0]);
      assertEquals(new String(takeResult[0]), threadTestString);
    }
  }

}
